package hasthi.common;

import hasthi.common.constants.HasthiConstants;
import hasthi.container.UnicastContianier;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public abstract class AbstractMessageBag<T> {
    protected static Log log = LogFactory.getLog(HasthiConstants.LOGGER_NAME);
    protected ConcurrentLinkedQueue<T> queue = new ConcurrentLinkedQueue<T>();
    protected AtomicInteger count = new AtomicInteger();
    protected UnicastContianier soapContainer;

    public AbstractMessageBag(UnicastContianier soapContainer) {
        this.soapContainer = soapContainer;
        new Thread(new Runnable() {
            public void run() {
                while (true) {
                    try {
                        int waitingCount = count.intValue();
                        
                        if (waitingCount > 20) {
                            if(waitingCount > 1000){
                                log.info(waitingCount + " Messages waiting to be processed ");
                            }
                            triggerBatchRemove(10);
                        } else if (waitingCount > 0) {
                            triggerBatchRemove(10);
                            Thread.sleep(5000);
                        } else {
                            Thread.sleep(5000);
                        }
                    } catch (Exception e) {
                        log.error(e);
                    }
                }

            }

        }).start();
    }

    public void add(T val) throws HasthiException {
        queue.add(val);
        count.incrementAndGet();
        if (count.get() > 10) {
            triggerBatchRemove(10);
        } 
    }

    private void triggerBatchRemove(int removeCount) throws HasthiException {
        T request;
        Set<T> removeset = new HashSet<T>(removeCount);
        while (removeset.size() < removeCount && (request = queue.poll()) != null) {
            removeset.add(request);
            this.count.decrementAndGet();
        }
        processBatch(removeset);
    }

    protected abstract void processBatch(Set<T> list) throws HasthiException;
    
    public int getSize(){
        return count.intValue();
    }
}
